package tableapi.udf

import bean.SensorReading
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.table.api.scala.{StreamTableEnvironment, _}
import org.apache.flink.table.api.{EnvironmentSettings, Table}
import org.apache.flink.table.functions.{AggregateFunction, ScalarFunction, TableFunction}
import org.apache.flink.types.Row

/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/11/28 23:10
  */
object Test3_AggregateFunction {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    //file source
    val inputPath: String = "src/main/resources/sensor.txt"
    val fileStream: DataStream[String] = env.readTextFile(inputPath)

    val socketStream = env.socketTextStream("192.168.88.106", 7777)

    //先转换为特定的类型
    val dataStream: DataStream[SensorReading] = fileStream.map(data => {
      val arr = data.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) {
        override def extractTimestamp(t: SensorReading): Long = t.timestamp * 1000L
      })

    val settings: EnvironmentSettings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()

    //创建表执行环境
    val tableEnv = StreamTableEnvironment.create(env, settings)

    val sensorTable: Table = tableEnv.fromDataStream(dataStream, 'id, 'temperature, 'timestamp.rowtime as 'ts)

    val avgTemp = new AvgTemp()
    val resultTabel:Table = sensorTable
        .groupBy('id)
        .aggregate(avgTemp('temperature) as 'avgTemp)
        .select('id,'avgTemp)

    //sql
    tableEnv.createTemporaryView("sensor",sensorTable)
    tableEnv.registerFunction("avgTemp",avgTemp)
    val resultSqlTable =tableEnv.sqlQuery(
      """
        |select id,avgTemp(temperature)
        |from
        |sensor
        |group by id
      """.stripMargin)

    resultTabel.toRetractStream[Row].print("result")
    resultSqlTable.toRetractStream[Row].print("sql")

    env.execute("aggregate test")


  }

  //定义一个类表示聚合状态
  class AvgTempAcc {
    var sum: Double = 0.0
    var count: Int = 0
  }

  //自定义一个聚合函数，求每个传感器的平均值
  class AvgTemp extends AggregateFunction[Double, AvgTempAcc] {
    override def getValue(acc: AvgTempAcc): Double = acc.sum / acc.count

    override def createAccumulator(): AvgTempAcc = new AvgTempAcc

    //实现一个具体的处理计算函数，accumulate
    def accumulate(accumulator: AvgTempAcc, temp: Double): Unit = {
      accumulator.sum += temp
      accumulator.count += 1

    }
  }


}
